package com.yequan.rocketmq.service.impl;

import com.alibaba.fastjson.JSON;
import com.yequan.rocketmq.service.RocketMQService;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/**
 * @author yq
 * @date 2020/09/10
 */
public class RocketMQSyncServiceImpl implements RocketMQService {

    private RocketMQProducer syncProducer;

    @Override
    public void sendMessage(String topic, String tag, byte[] body) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(topic, tag, body);
        syncProducer.getProducer().send(message);
    }

    @Override
    public void sendMessage(String topic, String tag, String msg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        Message message = new Message(topic, tag, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
        syncProducer.getProducer().send(message);
    }

    @Override
    public void sendMessage(String topic, String tag, Object msgObj) throws InterruptedException, RemotingException, MQClientException, MQBrokerException, UnsupportedEncodingException {
        Message message = new Message(topic, tag, JSON.toJSONString(msgObj).getBytes(RemotingHelper.DEFAULT_CHARSET));
        syncProducer.getProducer().send(message);
    }

    public void setSyncProducer(RocketMQProducer syncProducer) {
        this.syncProducer = syncProducer;
    }

    public RocketMQProducer getSyncProducer() {
        return syncProducer;
    }
}
